package sparkStream

import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.{SPARK_BRANCH, SparkConf}
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.{Seconds, StreamingContext}

object SparkKafka {
  def main(args: Array[String]): Unit = {
    /* spark streaming实现kafka 的消费者
    1) 构建 sparkconf本地运行，运行应用程序名称
    2)构建 sparkstreaming ---》streamingContext，加载配置
    3) kafka 配置 broker , key value , group id，消费模式
    4) spark 链接kafka 订阅，topic，streamingcontext
    5) 循环的形式打印/处理
    6) 开启ssc ，监控 kafka 数据*/

    // 1) 构建 sparkconf本地运行，运行应用程序名称
    val conf =new SparkConf().setMaster("local[*]").setAppName("helloSparkKafka")
    //sparkStream需导入依赖
    // spark streaming可以进行流式处理，微批次处理，间隔2秒
    val ssc=new StreamingContext(conf,Seconds(2))
    // spark 输出红色 info信息	--> error
    ssc.sparkContext.setLogLevel("error")
    // 4) spark 链接kafka 订阅，topic,streamingcontext// topic namel
    val kfkaParams=Map[String,Object](
      "bootstrap.Servers"->"192.168.121.129:9092",
      "key.deserializer"->classOf[StringDeserializer],
      "value.deserializer"->classOf[StringDeserializer],
      "group.id"->"niit",
      "enable,auto.commit"->(false:java.lang.Boolean)

    )
    val  topicName=Array("15test")
    // 返回kafka 返回的 streamRdd(一段有时间间隔的RDD)
    val streamRdd=KafkaUtils.createDirectStream[String,String](
      ssc,
      PreferConsistent,
      Subscribe[String,String](topicName,kfkaParams)
    )
    streamRdd.foreachRDD(
      x=>(
        if (!x.isEmpty()){
          val line=x.map(_.value())
          line.foreach(println)
        }
      )
    )
    //6) 开启ssc ，监控 kafka 数据
    ssc.start()
    ssc.awaitTermination()// 监控kafka的数据
  }

}
